package com.neusoft.databus.policy.listener;

import java.util.stream.Collectors;

import org.apache.tomcat.util.buf.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.bean.SnmptrapBean;
import com.neusoft.databus.policy.service.MessageServie;

import lombok.extern.slf4j.Slf4j;

/**
 * kafka listener. the only data entry of application
 *
 * @author naxiang
 */
@Slf4j
@Component
@ConditionalOnClass(KafkaTemplate.class)
public class KafkaMessageListener {
    @Autowired
    private MessageServie messageServie;

    /**
     * monitor collected data from kafka topic
     *
     * @param message
     */
    @KafkaListener(topics = "${databus.policy.monitorDataTopic}", group = "${spring.kafka.consumer.group-id}")
    public void processMessage(final MessageBean message) {
        final CommandBean commandBean = JsonUtils.jsonToPojo(message.getMessage(), CommandBean.class);
        KafkaMessageListener.log.info("receive a message from kafka, the resource code are: {} ", StringUtils
                .join(commandBean.getResourceBean().stream().map(it -> it.getCode()).collect(Collectors.toList())));
        try {
            this.messageServie.process(commandBean);
        } catch (final Exception e) {
            KafkaMessageListener.log.error("error occured when process kafka monitor topic message", e);
        }

    }

    @KafkaListener(topics = "${databus.policy.snmpTrapTopic}", group = "${spring.kafka.consumer.group-id}")
    public void processTrapMessage(final MessageBean message) {
        final SnmptrapBean snmptrapBean = JsonUtils.jsonToPojo(message.getMessage(), SnmptrapBean.class);
        KafkaMessageListener.log.info("receive a trap message from kafka, the resource id is: {} ",
                snmptrapBean.getResourceId());
        if (KafkaMessageListener.log.isDebugEnabled()) {
            KafkaMessageListener.log.debug("receive a message from kafka: {} ", snmptrapBean.toString());
        }
        try {
            this.messageServie.processTrapMessage(snmptrapBean);
        } catch (final Exception e) {
            KafkaMessageListener.log.error("error occured when process kafka trap topic message", e);
        }

    }
}
